import asyncio
import random
import re
import time
import math
import os
import ipaddress
import multiprocessing
from typing import List, Union, Callable


class ScannerCore:
    """核心配置类"""

    def __init__(self):
        self.chunk_size = 400
        self.cache_ttl = 300
        self.open_cache = True  # 缓存开关
        self.socket_timeout = 0.5
        self.timeout_multiplier = 1.5
        self.min_total_timeout = 10.0
        self.default_start = 11
        self.default_end = 29990
        self.verbose = False
        self.show_closed = False
        self.cache = {}
        self.on_message: Union[Callable[[str], None], None] = None
        self.is_only = False  # 是否仅一个端口


core = ScannerCore()


# ===============================
# 输出封装
# ===============================
def emit(msg: str):
    """统一输出，可回调或打印"""
    on_message = core.on_message
    if on_message:
        try:
            on_message(msg)
        except Exception as e:
            if core.verbose:
                print(f"[callback error] {e}")
        return
    if core.verbose:
        print(msg)


# ===============================
# 安全运行异步函数
# ===============================
def run_async(coro):
    try:
        return asyncio.run(coro)
    except RuntimeError:
        loop = asyncio.get_event_loop()
        return loop.run_until_complete(coro)


# ===============================
# 单端口检测
# ===============================
async def check_single_port(ip: str, port: int) -> bool:
    try:
        reader, writer = await asyncio.wait_for(
            asyncio.open_connection(ip, port),
            timeout=core.socket_timeout
        )
        writer.close()
        await writer.wait_closed()
        return True
    except Exception:
        return False


# ===============================
# 代理污染检测
# ===============================
async def check_proxy_pollution(ip: str) -> bool:
    """检测是否有代理或VPN污染"""
    old_verbose = core.verbose
    old_show_closed = core.show_closed
    old_on_message = core.on_message

    core.verbose = False
    core.show_closed = False
    core.on_message = None

    fake_ports = random.sample(  # 随机生成假端口
        [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 12, 14, 15, 16, 18, 24, 25, 26, 27, 28], 10)
    tasks = [check_single_port(ip, p) for p in fake_ports]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    core.verbose = old_verbose
    core.show_closed = old_show_closed
    core.on_message = old_on_message

    # 如果这些随机端口中存在能连通的，说明有代理污染
    return not any(isinstance(r, bool) and r for r in results)


# ===============================
# 异步扫描
# ===============================
async def scan_ports(ip: str, ports: List[int], limit: int, total_timeout: float) -> List[int]:
    semaphore = asyncio.Semaphore(limit)
    open_ports = []

    async def worker(port: int):
        async with semaphore:
            ok = await check_single_port(ip, port)
            if ok:
                emit(f"[+] port: {port:<5} True")
                return port
            elif core.show_closed:
                emit(f"[-] port: {port:<5} False")
            return None

    start_time = time.time()

    for i in range(0, len(ports), core.chunk_size):
        if time.time() - start_time > total_timeout:
            emit("[warn] 扫描超时，部分结果可能丢失")
            break

        batch = ports[i:i + core.chunk_size]
        tasks = [asyncio.create_task(worker(p)) for p in batch]

        done, pending = await asyncio.wait(tasks, timeout=total_timeout)
        for task in done:
            try:
                result = task.result()
                if result:
                    open_ports.append(result)
            except Exception:
                pass
        for task in pending:
            task.cancel()

    return sorted(open_ports)


# ===============================
# 输入解析与验证
# ===============================
def parse_target(target: str, start: int = None, end: int = None) -> tuple[str, List[int]]:
    """
    统一解析目标字符串:
    1. 支持中文符号、空格
    2. 自动验证 IP
    3. 支持多种端口格式: 80,443 或 80~100
    4. 支持 start/end 参数覆盖
    """
    if not target:
        raise ValueError(f"无效输入: {target}")

    # 每次调用前重置状态
    core.is_only = False

    # 清洗输入，兼容中英文符号与空格
    clean = (
        target.replace("，", ",")
        .replace("：", ":")
        .replace("。", ".")
        .replace(" ", "")
    )

    # 分离 IP 与端口部分
    if ":" in clean:
        ip, port_part = clean.split(":", 1)
    else:
        ip, port_part = clean, ""  # 没有端口时默认空字符串

    # 验证 IP 格式
    try:
        ipaddress.ip_address(ip)
    except ValueError:
        raise ValueError(f"无效IP地址: {ip}")

    # 检测是否被代理污染
    if not run_async(check_proxy_pollution(ip)):
        emit("[error] 检测到可能的代理污染，请关闭代理/VPN后重试。")
        return None

    # 参数覆盖规则
    if start is not None and end is not None:
        port_part = f"{start}~{end}"
    elif (start is not None) ^ (end is not None):
        raise ValueError("参数错误：start 与 end 必须同时指定")

    # 如果没有端口信息，启用默认扫描范围
    if not port_part:
        port_part = f"{core.default_start}~{core.default_end}"

    # 判断端口格式
    if ',' in port_part and '~' in port_part:
        raise ValueError("参数错误：不支持逗号与范围混用")

    # 多端口: 80,443,3306
    elif ',' in port_part:
        try:
            ports = [int(p) for p in port_part.split(",") if p]
        except ValueError:
            raise ValueError("参数错误：端口必须是整数")
        core.show_closed = True
        core.open_cache = False  # 即时查询，不缓存

    # 范围形式: 80~100
    elif '~' in port_part:
        try:
            left, right = map(int, port_part.split("~", 1))
        except ValueError:
            raise ValueError("参数错误：端口范围必须是整数")

        # 修正边界
        left = max(1, min(left, 65535))
        right = max(1, min(right, 65535))
        if left > right:
            left, right = right, left
        ports = range(left, right + 1)
        core.show_closed = False
        core.open_cache = True  # 范围扫描可缓存

    # 单端口
    else:
        try:
            port = int(port_part)
        except ValueError:
            raise ValueError("参数错误：端口必须是整数")
        ports = port
        core.show_closed = True
        core.open_cache = False
        core.is_only = True

    return ip, ports


# ===============================
# 主控制函数
# ===============================
def testIpStream(
    target: str,
    start: int = None,
    end: int = None,
    limit: int = 400,
    on_message: Callable[[str], None] = None
) -> Union[bool, List[int], None]:
    """
    主控制函数:
    - 统一调用 parse_target() 解析参数并检测代理污染
    - 支持缓存机制
    - 自动计算超时时间
    - 统一消息输出
    """
    core.on_message = on_message
    core.chunk_size = min(limit, multiprocessing.cpu_count() * 200)

    try:
        parse_result = parse_target(target, start, end)
        if not parse_result:
            return None
        ip, ports = parse_result

        # 单端口模式：直接返回 True/False
        if core.is_only:
            ok = run_async(check_single_port(ip, int(ports)))
            emit(f"[scan] [+] port: {ports:<5} {ok}")
            return ok

        # 统一端口列表格式
        port_list = list(ports) if not isinstance(ports, list) else ports

        # 缓存机制
        cache_key = f"{ip}_{','.join(map(str, port_list))}_{limit}"
        if core.open_cache:
            cache = core.cache.get(cache_key)
            if cache and time.time() - cache["time"] < core.cache_ttl:
                emit("[cache] 命中缓存")
                cached_ports = cache["data"]
                if cached_ports:
                    for p in cached_ports:
                        emit(f"[+] port: {p:<5} True")
                else:
                    emit("[cache] 无开放端口缓存记录")
                return cached_ports

        # 动态超时计算
        port_count = len(port_list)
        batches = math.ceil(port_count / limit)
        total_timeout = max(
            batches * core.socket_timeout * core.timeout_multiplier,
            core.min_total_timeout
        )
        emit(f"[info] 超时时间: {total_timeout:.1f}s")

        # 执行扫描
        result = run_async(scan_ports(ip, port_list, limit, total_timeout))

        # 写入缓存
        if core.open_cache:
            core.cache[cache_key] = {"time": time.time(), "data": result}

        return result

    except ValueError as e:
        emit(f"[error] 参数错误: {e}")
        return []
    except Exception as e:
        emit(f"[error] 执行异常: {e}")
        return []
